package com.company.bookstore.test;

import com.alibaba.fastjson.JSON;
import com.company.bookstore.entities.order_module.Order;
import com.company.bookstore.qo.order_module.OrderQo;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;

import java.io.UnsupportedEncodingException;
import java.util.List;

import static org.apache.rocketmq.client.producer.SendStatus.SEND_OK;

/***
 *
 * RocketMQ 消息队列封装
 *
 * DefaultMQProducer（group）===》生产者组 同一类Producer 消息的集合，如果发送的是事务消息，且原始生产者发送后故障，则经理人服务器会联系同组其他生产回溯或者提交数据。
 *
 * 消息返回类{@link SendResult}  {@link org.apache.rocketmq.client.producer.SendStatus}
 *
 * send 方法 同步（适合安全较高的操作） 异步（实时性较高） ，单向（不需要返回值，如日志）
 *
 * 同步 通过返回true false 判断发送信息是否抵达broker
 *
 * 异步 通过callBack获取返回信息 通过线程池创建用户线程
 *
 * 单向
 *
 *
 * https://github.com/apache/rocketmq/blob/master/docs/cn/client/java/API_Reference_DefaultMQProducer.md
 *
 * 1、Topic：消息主题，通过 Topic 对不同的业务消息进行分类。
 *
 * 2、Tag：消息标签，用来进一步区分某个 Topic 下的消息分类，消息队列 RocketMQ 允许消费者按照 Tag 对消息进行过滤，确保消费者最终只消费到他关注的消息类型。
 *
 * Topic 与 Tag 都是业务上用来归类的标识，区分在于 Topic 是一级分类，而 Tag 可以说是二级分类，关系如图所示
 *
 * */
@Component
@Slf4j
public class CustomProducer implements InitializingBean {
    private DefaultMQProducer producer;

    @Override
    public void afterPropertiesSet() throws Exception {
        producer = new DefaultMQProducer("group_second_kill");
        producer.setNamesrvAddr("101.37.33.148:9876");
        producer.start();
    }
//    static {
////        producer.setNamesrvAddr("192.168.58.128:9876");
//          producer.setNamesrvAddr("101.37.33.148:9876");
//
////        //Launch the instance.
////        try {
////            producer.start();
////        } catch (MQClientException e) {
////            e.printStackTrace();
////        }
//    }
    public  boolean syncSend(List<Order> contents){
        // Specify name server addresses.

        //Create a message instance, specifying topic, tag and message body.
        try {
            for (Order content:contents
            ) {
                Message msg = new Message("topic_second_kill" /* Topic */,
                        "book" /* Tag */,
//                        (content).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
                        JSON.toJSONBytes(content)
                );
                SendResult sendResult = producer.send(msg);
                if (!sendResult.getSendStatus().equals(SEND_OK))
                {
                    return false;
                }
//                log.info( "syncSend ==>  {}", sendResult);
//                log.info("syncSend max offset {}",producer.maxOffset(sendResult.getMessageQueue()));
                //Shut down once the producer instance is not longer in use.
            }

            log.info("Method : syncSend =>发送成功");
            return true;
        }catch (RemotingException | InterruptedException | MQBrokerException | MQClientException u) {
            u.printStackTrace();
            return false;
        }
    }
//
//    public boolean asyncSend(List<String> contents){
//        try {
//            producer.setRetryTimesWhenSendAsyncFailed(0);
//            final CountDownLatch countDownLatch = new CountDownLatch(contents.size());
////            为什么要用final
////            当变量是final时，其内存空间在方法区，在多线程之间共享，所以多线程可见
////            为什么要用数组！
////            boolean 被定义为常量，当复制到内部类后，其依然是常量，不允许被赋值
////            其返回时 是否能作为判断依据
////            如果没被销毁，只要能调用，则可以作为判断依据
////
//             final boolean []flag = {true};
//            for (String content:contents
//            ) {
//                Message msg = new Message("topic-test" /* Topic */,
//                        "tag-test" /* Tag */,
//                        (content).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
//                );
//
//                producer.send(msg, new SendCallback() {
//                    @Override
//                    public void onSuccess(SendResult sendResult) {
//                        countDownLatch.countDown();
//                        flag[0] = false;
//                        log.info( "asyncSend ==>  {}", sendResult);
//                    }
//
//                    @Override
//                    public void onException(Throwable e) {
////                        出错没有具体的出错信息，与出错位置 待改善， 不同的业务如何实现不同错误逻辑实现
//                        log.error( "get Error! ==> {}",e,e);
//                        flag[0] = false;
//                        e.printStackTrace();
//                    }
//                });
//
//                //Shut down once the producer instance is not longer in use.
//            }
//
////            等待10秒 直到其他线程都执行完毕
//            countDownLatch.await(10, TimeUnit.SECONDS);
//
//            return flag[0];
//        }catch (UnsupportedEncodingException | RemotingException | InterruptedException | MQClientException u){
//            u.printStackTrace();
//            return false;
//        }
//    }


}
